package com.gongfutrip.trans.client;


import com.gongfutrip.trans.Producer;
import com.gongfutrip.trans.client.common.ClientUtils;
import com.gongfutrip.trans.client.common.CommunicationMode;
import com.gongfutrip.trans.client.common.Message;
import com.gongfutrip.trans.client.common.ProducerConfig;
import com.gongfutrip.trans.client.common.SendCallback;
import com.gongfutrip.trans.client.common.SendMessageRequestHeader;
import com.gongfutrip.trans.client.common.SendResult;
import com.gongfutrip.trans.client.common.SendResult.SendStatus;
import com.gongfutrip.trans.client.common.ServiceState;
import com.gongfutrip.trans.client.exception.ClientException;
import com.gongfutrip.trans.remoting.ChannelEventListener;
import com.gongfutrip.trans.remoting.InvokeCallback;
import com.gongfutrip.trans.remoting.NettyRemotingClient;
import com.gongfutrip.trans.remoting.RemotingClient;
import com.gongfutrip.trans.remoting.expception.RemotingCommandException;
import com.gongfutrip.trans.remoting.expception.RemotingConnectException;
import com.gongfutrip.trans.remoting.expception.RemotingException;
import com.gongfutrip.trans.remoting.expception.RemotingSendRequestException;
import com.gongfutrip.trans.remoting.expception.RemotingTimeoutException;
import com.gongfutrip.trans.remoting.expception.RemotingTooMuchRequestException;
import com.gongfutrip.trans.remoting.netty.NettyClientConfig;
import com.gongfutrip.trans.remoting.netty.ResponseFuture;
import com.gongfutrip.trans.remoting.protocol.RemotingCommand;
import com.gongfutrip.trans.remoting.protocol.RemotingSysResponseCode;
import java.io.UnsupportedEncodingException;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class ProducerImpl implements Producer {

    private static final Logger logger = LoggerFactory.getLogger(ProducerImpl.class);

    private ServiceState serviceState = ServiceState.CREATE_JUST;
    private final RemotingClient remotingClient;
    private final ProducerConfig producerConfig;

    public ProducerImpl(final ProducerConfig producerConfig,
        final ChannelEventListener channelEventListener) {
        this.remotingClient = new NettyRemotingClient(new NettyClientConfig(),
            channelEventListener);
        this.producerConfig = producerConfig;
        try {
            logger.info("#################### begin start producer ####################");
            this.start();
            logger.info("#################### start producer success ####################");
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void start() throws ClientException {
        switch (this.serviceState) {
            case CREATE_JUST:
                this.remotingClient.start();
                this.serviceState = ServiceState.RUNNING;
                break;
            case RUNNING:
            case START_FAILED:
            case SHUTDOWN_ALREADY:
                throw new ClientException(
                    "The producer service state not OK, maybe started once, " + this.serviceState);
            default:
                break;
        }

    }


    private SendResult sendMessage(Message msg, CommunicationMode communicationMode,
        SendCallback sendCallback, long timeoutMillis) throws RemotingException {
        try {
            setUniqID(msg);
            int sysFlag = 0;
            if (ClientUtils.tryToCompressMessage(msg)) {// 根据压缩配置执行消息压缩。
                sysFlag |= ClientUtils.COMPRESSED_FLAG;
            }

            this.remotingClient
                .updateNameServerAddressList(ProducerConfig.getNamesrvAddrList());// 将服务器地址更新到客户端
            // 组装发送消息的请求头。
            SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
            requestHeader.setSysFlag(sysFlag);// 压缩标识
            requestHeader.setBornTimestamp(System.currentTimeMillis());
            requestHeader.setProperties(ClientUtils.messageProperties2String(msg.getProperties()));
            requestHeader.setAddr(msg.getAddr());
            RemotingCommand request =
                RemotingCommand.createRequestCommand(msg.getRequestCode(), requestHeader);
            request.setBody(msg.getBody());
            // 三种通讯模型，调用对应的方法处理。
            switch (communicationMode) {
                case ONEWAY:
                    this.sendMessageOneWay(msg.getAddr(), request, timeoutMillis);
                    return null;
                case ASYNC:
                    this.sendMessageAsync(msg.getAddr(), msg, request, timeoutMillis,
                        producerConfig.getRetryTimesWhenSendAsyncFailed(), sendCallback);
                    return null;
                case SYNC:
                    return this.sendMessageSync(msg.getAddr(), msg, timeoutMillis, request,
                        producerConfig.getRetryTimesWhenSendFailed());
                default:
                    assert false;
                    break;
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    private void sendMessageAsync(String addr, Message msg, RemotingCommand request,
        long timeoutMillis, int retryTimesWhenSendAsyncFailed, SendCallback sendCallback) {
        int trytime = retryTimesWhenSendAsyncFailed;
        while (trytime > 0) {// 失败重试
            try {
                this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() {
                    /**
                     * 表示已经完成向服务端发送请求操作
                     */
                    public void operationComplete(ResponseFuture responseFuture) {
                        RemotingCommand response = responseFuture.getResponseCommand();
                        if (null != sendCallback && response != null) {
                            try {
                                SendResult sendResult = processSendResponse(msg, response);
                                try {
                                    sendCallback.onSuccess(sendResult);
                                } catch (Exception e) {
                                }
                            } catch (Throwable e) {
                                sendCallback.onException(e);
                            }
                        }
                    }
                });
                break;
            } catch (RemotingConnectException | RemotingTooMuchRequestException | RemotingTimeoutException
                | RemotingSendRequestException | InterruptedException e) {
                trytime--;
            }
        }

    }

    private void sendMessageOneWay(final String addr, final RemotingCommand request,
        final long timeoutMillis) {
        try {
            this.remotingClient.invokeOneway(addr, request, timeoutMillis);
        } catch (InterruptedException | RemotingConnectException | RemotingTooMuchRequestException | RemotingTimeoutException | RemotingSendRequestException e) {
            e.printStackTrace();
        }
    }

    private SendResult sendMessageSync(String addr, Message msg, long timeoutMillis,
        RemotingCommand request, int retryTimesWhenSendFailed)
        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
        InterruptedException, RemotingCommandException {
        SendResult sendResult = null;
        for (int i = 0; i < retryTimesWhenSendFailed; i++) {
            RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
            sendResult = processSendResponse(msg, response);
            if (null != sendResult && sendResult.getSendStatus() == SendStatus.SEND_OK) {
                break;
            }
        }
        return sendResult;
    }

    private SendResult processSendResponse(final Message msg, final RemotingCommand response)
        throws RemotingCommandException {
        SendResult sendResult = null;
        switch (response.getCode()) {
            case RemotingSysResponseCode.SUCCESS: {
                SendStatus sendStatus = SendStatus.SEND_OK;
                sendResult = new SendResult(sendStatus, getUniqID(msg), response.getBody());
                break;
            }
            case RemotingSysResponseCode.SYSTEM_ERROR: {
                SendStatus sendStatus = SendStatus.SYSTEM_ERROR;
                sendResult = new SendResult(sendStatus);
                break;
            }
            case RemotingSysResponseCode.SYSTEM_BUSY: {
                SendStatus sendStatus = SendStatus.SYSTEM_BUSY;
                sendResult = new SendResult(sendStatus);
                break;
            }
            case RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED: {
                SendStatus sendStatus = SendStatus.REQUEST_CODE_NOT_SUPPORTED;
                sendResult = new SendResult(sendStatus);
                break;
            }
            default:
                SendStatus sendStatus = SendStatus.SYSTEM_ERROR;
                sendResult = new SendResult(sendStatus);
                break;
        }
        return sendResult;
    }


    private void setUniqID(final Message msg) {
        if (msg.getProperty("UNIQ_KEY") == null) {
            msg.putProperty("UNIQ_KEY", createUniqID());
        }
    }

    public static String getUniqID(final Message msg) {
        return msg.getProperty("UNIQ_KEY");
    }

    private String createUniqID() {
        return UUID.randomUUID().toString().replaceAll("-", "").toUpperCase();
    }

    public SendResult sendSync(Message msg, long timeout) throws ClientException {
        return this.send(msg, CommunicationMode.SYNC, null, timeout);
    }

    public SendResult sendSync(Message msg) throws ClientException {
        return this.send(msg, CommunicationMode.SYNC, null, producerConfig.getSendMsgTimeout());
    }

    public SendResult sendAsync(Message msg, long timeout, SendCallback sendCallback)
        throws ClientException {
        return this.send(msg, CommunicationMode.ASYNC, sendCallback, timeout);
    }

    public SendResult sendAsync(Message msg, SendCallback sendCallback) throws ClientException {
        return this.send(msg, CommunicationMode.ASYNC, sendCallback,
            producerConfig.getSendMsgTimeout());
    }

    public SendResult sendOneWay(Message msg, long timeout) throws ClientException {
        return this.send(msg, CommunicationMode.ONEWAY, null, timeout);
    }

    public SendResult sendOneWay(Message msg) throws ClientException {
        return this.send(msg, CommunicationMode.ONEWAY, null, producerConfig.getSendMsgTimeout());
    }

    public SendResult send(Message msg, final CommunicationMode communicationMode,
        final SendCallback sendCallback, final long timeout) throws ClientException {
        makeSureStateOK();
        checkMessage(msg);
        long currentTime = System.currentTimeMillis();
        long beginTimestampPrev = currentTime;// 发送开始时间。
        long endTimestamp = currentTime;
        SendResult sendResult = null;
        beginTimestampPrev = System.currentTimeMillis();
        try {
            sendResult = this
                .sendMessage(msg, communicationMode, sendCallback, timeout);// 调用该方法真正发送。
            endTimestamp = System.currentTimeMillis();// 记录发送耗时
            logger.info("send message tasks time :" + (endTimestamp - beginTimestampPrev) + " ms  "
                + communicationMode);
            switch (communicationMode) {
                case ASYNC:
                    return null;// 异步发送立即返回。
                case ONEWAY:
                    return null;// 单向通讯也立即返回。
                case SYNC:// 同步发送则如果未成功则重试，否则直接返回发送结果。
                    return sendResult;
                default:
                    return null;
            }
        } catch (RemotingException e) {
            logger.debug(e.getMessage());
        }
        return sendResult;
    }

    public void checkMessage(Message msg) throws ClientException {
        if (null == msg) {
            throw new ClientException("the message is null");
        }
        if (null == msg.getBody()) {
            throw new ClientException("the message body is null");
        }
        if (0 == msg.getBody().length) {
            throw new ClientException("the message body length is zero");
        }
        if (msg.getBody().length > producerConfig.getMaxMessageSize()) {
            throw new ClientException(
                "the message body size over max value, MAX: " + producerConfig.getMaxMessageSize());
        }
    }

    private void makeSureStateOK() throws ClientException {
        if (this.serviceState != ServiceState.RUNNING) {
            throw new ClientException("The producer service state not OK, " + this.serviceState);
        }
    }


    @Override
    public void shutdown() {
        this.remotingClient.shutdown();
        this.serviceState = ServiceState.SHUTDOWN_ALREADY;
    }


    public static void main(String[] args)
        throws UnsupportedEncodingException, ClientException, InterruptedException {

        // String addr = "52.79.138.27";
        String addr = "127.0.0.1";
        ProducerImpl producer = new ProducerImpl(new ProducerConfig(), null /*new ChannelEventListener() {
      @Override
      public void onChannelConnect(String remoteAddr, Channel channel) {
        System.out.println("onChannelConnect: "+remoteAddr+"-->"+channel);
      }

      @Override
      public void onChannelClose(String remoteAddr, Channel channel) {
        System.out.println("onChannelClose: "+remoteAddr+"-->"+channel);
      }

      @Override
      public void onChannelException(String remoteAddr, Channel channel) {
        System.out.println("onChannelException: "+remoteAddr+"-->"+channel);
      }

      @Override
      public void onChannelIdle(String remoteAddr, Channel channel) {
        System.out.println("onChannelIdle: "+remoteAddr+"-->"+channel);
      }
    }*/);
        for (int i = 0; i < 100; i++) {
            Message msg = new Message(BusinesssRequestCode.SHOPPINBG, addr + ":8888",
                ("{\"buyerGTPId\":16,\"cacheTime\":2700000,\"groupId\":49,\"roleId\":0,\"sourceType\":\"API\",\"timeout\":20000,\"token\":\"84108111-10c6-49af-b6b4-3a60a0247810\",\"uid\":\"RksVSX7PfZm1yF04adBWYsCD7M4=\",\"useLcc\":false,\"useTeam\":false},{\"buyerCurrency\":\"CNY\",\"dataSourceGTPId\":1024,\"extraInfo\":{\"productCode\":\"TS1700014632\",\"productSource\":\"UTOUR\",\"searchKey\":\"BJSBKK20170509YBKKBJS20170514Y1000\",\"solutionKey\":\"0c7dda08b5df4ac4854c539484e05652\"},\"flights\":[{\"airline\":\"TG\",\"arrivalDate\":\"2017-05-09\",\"arrivalTime\":\"21:15\",\"baseSegments\":[{\"airline\":\"TG\",\"arrival\":\"BKK\",\"arrivalDate\":\"2017-05-09\",\"arrivalTime\":\"21:15\",\"arrivalTimestamp\":1494335700000,\"availabilityCount\":1,\"bookingCode\":\"G\",\"bookingCodeMap\":{\"G\":1},\"cabinClass\":\"ECONOMY\",\"cabinCodeMap\":{},\"changePlane\":false,\"codeShare\":\"N\",\"connection\":false,\"departure\":\"PEK\",\"departureDate\":\"2017-05-09\",\"departureTime\":\"17:05\",\"equipment\":\"773\",\"extInfo\":{},\"fareBasis\":\"YRT\",\"flightNum\":\"615\",\"flightSeq\":\"0\",\"flightTime\":310,\"index\":0,\"lowerBookingCodeMap\":{},\"opFltAirline\":\"\",\"opFltNo\":\"\",\"seqNum\":0,\"terminalDep\":\"3\"}],\"bookingCodeKey\":\"G\",\"departureDate\":\"2017-05-09\",\"departureTime\":\"17:05\",\"destination\":\"BKK\",\"flightType\":\"RT\",\"gds\":\"TEAM\",\"origin\":\"BJS\",\"ruleRefs\":[],\"segmentKey\":\"TG615\",\"segmentType\":\"1\",\"travelTime\":310},{\"airline\":\"TG\",\"arrivalDate\":\"2017-05-14\",\"arrivalTime\":\"15:50\",\"baseSegments\":[{\"airline\":\"TG\",\"arrival\":\"PEK\",\"arrivalDate\":\"2017-05-14\",\"arrivalTime\":\"15:50\",\"arrivalTimestamp\":1494748200000,\"availabilityCount\":1,\"bookingCode\":\"G\",\"bookingCodeMap\":{\"G\":1},\"cabinClass\":\"ECONOMY\",\"cabinCodeMap\":{},\"changePlane\":false,\"codeShare\":\"N\",\"connection\":false,\"departure\":\"BKK\",\"departureDate\":\"2017-05-14\",\"departureTime\":\"10:10\",\"equipment\":\"773\",\"extInfo\":{},\"fareBasis\":\"YRT\",\"flightNum\":\"614\",\"flightSeq\":\"0\",\"flightTime\":280,\"index\":0,\"lowerBookingCodeMap\":{},\"opFltAirline\":\"\",\"opFltNo\":\"\",\"seqNum\":0,\"terminalArr\":\"3\"}],\"bookingCodeKey\":\"G\",\"departureDate\":\"2017-05-14\",\"departureTime\":\"10:10\",\"destination\":\"BJS\",\"flightType\":\"RT\",\"gds\":\"TEAM\",\"origin\":\"BKK\",\"ruleRefs\":[],\"segmentKey\":\"TG614\",\"segmentType\":\"2\",\"travelTime\":280}],\"hasPriced\":false,\"latestTktTime\":\"2017-05-04 17:05:00\",\"passengers\":[\"ADT\"],\"platingCarrier\":\"TG\",\"prices\":{\"ADT\":{\"buyerCurrency\":\"CNY\",\"buyerDistributionFee\":0,\"buyerMarkup\":0,\"buyerOtherCharge\":0,\"buyerPlatformServiceFee\":0,\"buyerPriceAdjustment\":0,\"buyerQcharge\":0,\"buyerServiceFee\":0,\"buyerTKTFee\":0.0,\"buyerTaxes\":250.0,\"buyerTicketPrice\":2650.0,\"buyerTotalPrice\":2900.0,\"footmark\":[{\"key\":\"step1.1\",\"value\":\"Fare LCF exchange: 1325 x 1 = 1325<br>Q LCF exchange: 0 x 1 = 0<br>Total LCF combine: 1325 + 0 = 1325\"},{\"key\":\"step1.2\",\"value\":\"Fare LCF exchange: 1325 x 1 = 1325<br>Q LCF exchange: 0 x 1 = 0<br>Total LCF combine: 1325 + 0 = 1325\"},{\"key\":\"step2\",\"value\":\"merge ( TLCF_?{1:1325, 2:1325} ) =  TLCF{2650}<br>round_fare ( TLCF{2650} ,  POC{CNY} ,  REGION{Worldwide} ,  COUNTRY{CN} ) =  TLCF{2650}<br>exchange_gds ( POC{CNY} ,  POS{CNY} ,  GDS{1T} ,  GTP{GTP} ) =  EXC_GDS{1}<br>TLCF{2650}  *  EXC_GDS{1}  =  BFARE{2650}<br>round_fare ( BFARE{2650} ,  POS{CNY} ,  REGION{Worldwide} ,  COUNTRY{CN} ) =  SUPPLIER_FARE{2650}<br>merge ( TAX_DETAIL_?{PK:250} ) =  SUPPLIER_TAX{250}<br>Supplier price merge: supplierFare:2650 + supplierTax:250 + supplierTKTFee:0 = 2900\"},{\"key\":\"step3\",\"value\":\"Fare s2b exchange: 2650 x 1 >> 2650<br>Tax s2b exchange: 250 x 1 >> 250<br>TktFee s2b exchange: 0 x 1 >> 0<br>Fare buyer merge: buyerFare:2650 + buyerTaxes:250 + buyerTKTFee:0 = 2900\"},{\"key\":\"farebasis\",\"value\":\"YRT\"},{\"key\":\"basic\",\"value\":\"fc:null<br>(p)p2s.GDS:CNY>CNY:1T/2017-04-18/1<br>s2b.bank:CNY>CNY:null/2017-04-18 19:51:37.3/1<br>roe:1<br>s.taxes:250<br>PK:250<br>nuc1.fare:1325<br>nuc1.Q:0<br>nuc2.fare:1325<br>nuc2.Q:0<br>\"},{\"key\":\"policy\",\"value\":\"\"},{\"key\":\"datasource\",\"value\":\"team\"},{\"key\":\"tktpcc\",\"value\":\"team\"},{\"key\":\"fareSource\"},{\"key\":\"supplierCode\",\"value\":\"BJS-TEAM-1T\"},{\"key\":\"markup\",\"value\":\"----\"}],\"p2sGDSRate\":1,\"priceCalculation\":{\"nUCBasePrices\":[1325.00,1325.00],\"nUCQCharges\":[0,0],\"nUCTotalPrice\":2900.00,\"pOCCurrency\":\"CNY\",\"rOE\":1,\"supplierDetailTaxes\":{},\"supplierFaceValue\":2650.00},\"s2bBankRate\":1,\"supplierCurrency\":\"CNY\",\"supplierFaceTaxValue\":250.00,\"supplierFaceValue\":2650.00,\"supplierServiceFee\":0,\"supplierTKTFee\":0,\"supplierTaxes\":250.00,\"supplierTicketPrice\":2650.0,\"supplierTotalPrice\":2900.00,\"taxMap\":{\"PK\":250.00}}},\"tKTSupplierGTPId\":1024},{\"needAVGTPIds\":[1024],\"needPricingGTPIds\":[1024]},\"ALL\""
                    + i).getBytes("utf-8"));
            // SendResult sendResult = producer.sendSync(msg);
            // logger.info(sendResult.toString());
            //
            producer.sendAsync(msg, new SendCallback() {

                @Override
                public void onSuccess(SendResult sendResult) {
                    logger.info("发送给服务端完成收到结果：" + sendResult.toString());
                }

                @Override
                public void onException(Throwable e) {
                    logger.info(e.getMessage());
                }
            });
        }
    }
}
